package cn.doitedu.api;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/27
 * @Desc: 学大数据，上多易教育
 *    全局计数滚动窗口api示例
 **/
public class _21_GlobalWindow_Count_Tumbling_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        // 先把字符串转成整数
        SingleOutputStreamOperator<Integer> mapped = stream.map(Integer::parseInt);

        // 我们要按每10条元素来计算元素之和
        SingleOutputStreamOperator<Integer> mainOutput =
                mapped
                        .countWindowAll(5)
                        .process(new ProcessAllWindowFunction<Integer, Integer, GlobalWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<Integer, Integer, GlobalWindow>.Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {

                                int sum = 0;
                                for (Integer element : elements) {

                                    // 如果是奇数，则输出到一个侧流中
                                    if (element % 2 == 1) {
                                        context.output(new OutputTag<>("odd-digits", TypeInformation.of(Integer.class)), element);
                                    } else {
                                        // 如果是偶数，则进行求和
                                        sum += element;
                                    }
                                }

                                out.collect(sum);
                            }
                        });


        SideOutputDataStream<Integer> sideOutput = mainOutput.getSideOutput(new OutputTag<>("odd-digits", TypeInformation.of(Integer.class)));

        mainOutput.print("main >>> ");
        sideOutput.print("side >>> ");


        env.execute();


    }
}
